#include "rpcprovider.h"
#include "mprpcapplication.h"
#include "rpcheader.pb.h"
#include "logger.h"
#include "zookeeperutil.h"
#include <iostream>

void RpcProvider::NotifyService(google::protobuf::Service *service)
{
    ServiceInfo service_info; // 结构体

    // 获取了服务对象的描述信息
    const google::protobuf::ServiceDescriptor *pserviceDesc = service->GetDescriptor();
    // 因为返回类型是指针。获取服务对象的描述信息。存储名字之类的。

    // 获取服务的名字
    std::string service_name = pserviceDesc->name();
    // 获取服务对象service的方法的数量
    int methodCnt = pserviceDesc->method_count();

    // std::cout << "service_name:" << service_name << std::endl;
    LOG_INFO("service_name:%s", service_name.c_str());

    for (int i = 0; i < methodCnt; ++i)
    {
        // 获取了服务对象指定下标的服务方法的描述（抽象的描述） UserService   Login
        const google::protobuf::MethodDescriptor *pmethodDesc = pserviceDesc->method(i);
        std::string method_name = pmethodDesc->name();
        service_info.m_methodMap.insert({method_name, pmethodDesc}); // 插入键值对到map中

        // std::cout<<"method_name:"<<method_name<<std::endl;//打印
        LOG_INFO("method_name:%s", method_name.c_str());
    }
    service_info.m_service = service;                  // 记录服务对象
    m_serviceMap.insert({service_name, service_info}); // 存储一下服务及其具体的描述
}

void RpcProvider::Run()
{
    // 读配置文件，获取连接信息
    std::string ip = MprpcApplication::GetInstance().GetConfig().Load("rpcserverip");
    uint16_t port = atoi(MprpcApplication::GetInstance().GetConfig().Load("rpcserverport").c_str());
    muduo::net::InetAddress address(ip, port);

    // 创建TcpServer对象
    muduo::net::TcpServer server(&m_eventLoop, address, "RpcProvider");
    // 绑定连接回调和消息读写回调方法
    auto connect_callback = [this](const muduo::net::TcpConnectionPtr &conn)
    {
        return this->OnConnection(conn);
    };
    auto message_callback = [this](const muduo::net::TcpConnectionPtr &conn, muduo::net::Buffer *buffer, muduo::Timestamp time)
    {
        return this->OnMessage(conn, buffer, time);
    };
    server.setConnectionCallback(connect_callback);
    server.setMessageCallback(message_callback);

    // 设置muduo库的线程数量
    server.setThreadNum(4); // 1个是I/O线程，3个是工作线程

    // 把当前rpc节点上要发布的服务注册到zk上
    // session timeout   30s  zkclient 网络I/O线程  1/3 * timeout 时间发送ping消息
    ZkClient zkCli;
    zkCli.Start();
    // service_name为永久性节点   method_name为临时性节点
    for (auto &sp : m_serviceMap)
    {
        std::string service_path = "/" + sp.first;
        zkCli.Create(service_path.c_str(), nullptr, 0);
        for (auto &mp : sp.second.m_methodMap)
        {
            // method_path的一个可能值: /service_name/method_name   /UserServiceRpc/Login
            // method_path_data存储当前这个rpc服务节点主机的ip和port
            std::string method_path = service_path + "/" + mp.first;
            char method_path_data[128] = {0};
            sprintf(method_path_data, "%s:%d", ip.c_str(), port);
            // ZOO_EPHEMERAL表示znode是一个临时性节点，和zkserver断了，就是表示不提供这个RPC服务了，所以ZK自动删掉就好啦。
            zkCli.Create(method_path.c_str(), method_path_data, strlen(method_path_data), ZOO_EPHEMERAL);
        }
    }

    // rpc服务端准备启动，打印信息
    std::cout << "RpcProvider start service at ip:" << ip << " port:" << port << std::endl;

    // 启动网络服务
    server.start();
    m_eventLoop.loop(); // 相当于启动了epoll_wait，阻塞，等待远程连接
}

// 链接回调
void RpcProvider::OnConnection(const muduo::net::TcpConnectionPtr &conn)
{
    if (!conn->connected())
    {
        // 连接断开
        conn->shutdown(); // 关闭文件描述符
    }
}
// IO回调
void RpcProvider::OnMessage(const muduo::net::TcpConnectionPtr &conn,
                            muduo::net::Buffer *buffer,
                            muduo::Timestamp)
{
    // 网络上接收的远程rpc调用请求的字符流 包含了RPC方法的名字Login和参数args
    std::string recv_buf = buffer->retrieveAllAsString();

    // header_size 报头长度
    uint32_t header_size = 0;
    recv_buf.copy((char *)&header_size, 4, 0);

    std::string rpc_header_str = recv_buf.substr(4, header_size);
    // 包含service_name method_name args_size
    mprpc::RpcHeader rpcHeader;
    // 根据header_size读取数据头的原始字符流，反序列化数据，得到rpc请求的详细信息
    std::string service_name;
    std::string method_name;
    uint32_t args_size;
    if (rpcHeader.ParseFromString(rpc_header_str))
    {
        // 数据头反序列化成功
        service_name = rpcHeader.service_name();
        method_name = rpcHeader.method_name();
        args_size = rpcHeader.args_size();
    }
    else
    {
        // 数据头反序列化失败
        std::cout << "rpc_header_str:" << rpc_header_str << " parse error!" << std::endl;
        return; // 不用往后走了
    }

    // 获取rpc方法参数的字符流数据
    std::string args_str = recv_buf.substr(4 + header_size, args_size);
    // header_size(4个字节) + header_str + args_str

    // 打印调试信息
    std::cout << "============================================" << std::endl;
    std::cout << "header_size: " << header_size << std::endl;
    std::cout << "rpc_header_str: " << rpc_header_str << std::endl;
    std::cout << "service_name: " << service_name << std::endl;
    std::cout << "method_name: " << method_name << std::endl;
    std::cout << "args_str: " << args_str << std::endl;
    std::cout << "============================================" << std::endl;

    // 查找对应服务
    auto it = m_serviceMap.find(service_name); // 用[]会有副作用
    if (it == m_serviceMap.end())              // 根本没有的服务
    {
        std::cout << service_name << " is not exist!" << std::endl;
        return;
    }

    // 查找对应方法
    auto mit = it->second.m_methodMap.find(method_name);
    if (mit == it->second.m_methodMap.end()) // 服务里没有这个方法
    {
        std::cout << service_name << ":" << method_name << " is not exist!" << std::endl;
        return;
    }

    google::protobuf::Service *service = it->second.m_service;      // 获取service对象  对应new UserService这种
    const google::protobuf::MethodDescriptor *method = mit->second; // 获取method对象 对应Login这种

    // 生成rpc方法调用的请求request和响应response参数
    google::protobuf::Message *request = service->GetRequestPrototype(method).New();
    // 在框架以抽象的方式表示。new生成新对象，传给userservice

    if (!request->ParseFromString(args_str)) // 解析
    {
        std::cout << "request parse error, content:" << args_str << std::endl;
        return;
    }
    google::protobuf::Message *response = service->GetResponsePrototype(method).New(); // 响应

    // CallMethod需要closure参数
    // 给下面的method方法的调用，绑定一个Closure的回调函数
    google::protobuf::Closure *done = google::protobuf::NewCallback<RpcProvider,
                                                                    const muduo::net::TcpConnectionPtr &,
                                                                    google::protobuf::Message *>(this,
                                                                                                 &RpcProvider::SendRpcResponse,
                                                                                                 conn, response);

    // 在框架上根据远端rpc请求，调用当前rpc节点上发布的方法

    service->CallMethod(method, nullptr, request, response, done); // 做完本地业务，根据结果写好reponse给框架，框架再给调用方
    // 相当于new UserService().Login(controller, request, response, done)
}

// Closure的回调操作，用于序列化rpc的响应和网络发送
void RpcProvider::SendRpcResponse(const muduo::net::TcpConnectionPtr &conn, google::protobuf::Message *response)
{
    std::string resonse_str;
    if (response->SerializeToString(&resonse_str))
    {
        conn->send(resonse_str);
    }
    else
    {
        std::cout << "serialize response_str error!" << std::endl;
    }

    conn->shutdown(); // 模拟http的短链接服务，由rpcprovider主动断开连接，给更多的rpc调用方提供服务
}
